package com.lh.lepay.controller;

import com.lh.lepay.entity.Chathistory;
import com.lh.lepay.entity.dto.ChatMessageHistoryDto;
import com.lh.lepay.service.ChatMemoryService;
import com.lh.lepay.service.ChathistoryService;
import com.lh.lepay.utils.ResponseData;
import dev.langchain4j.model.chat.ChatLanguageModel;
import dev.langchain4j.model.chat.StreamingChatLanguageModel;
import dev.langchain4j.model.chat.response.ChatResponse;
import dev.langchain4j.model.chat.response.StreamingChatResponseHandler;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
/**
 * (Chathistory)表控制层
 *
 * @author makejava
 * @since 2025-03-25 14:16:30
 */
@RestController
@RequiredArgsConstructor
@Slf4j
@RequestMapping("/ai")
public class ChatHistoryController {

    //private final ChatClient chatClient;
    private final ChatLanguageModel chatLanguageModel;
    private final StreamingChatLanguageModel streamingChatLanguageModel;
    // 线程池用于异步处理SSE
    private final ExecutorService executor = Executors.newCachedThreadPool();

    @Autowired
    private ChatMemoryService chatMemoryService;

//    public ChatController(ChatMemoryService chatMemoryService) {
//        this.chatMemoryService = chatMemoryService;
//    }

    @Resource
    private ChathistoryService chathistoryService;


    @PostMapping("/chatHistory")
    public ResponseData insertChatHistory(@RequestBody Chathistory chathistory) {
        ResponseData responseData = chathistoryService.insertChatHistoryStatment(chathistory);
        return responseData;
    }

//    @PostMapping("/chatHistory")
//    public ResponseData insertChatHistoryDto(@RequestBody ChatHistoryDto chatHistoryDto){
//        ResponseData responseData = chathistoryService.insertChatHistoryDto(chatHistoryDto);
//        return  responseData;
//    }

    @PostMapping("/getChatHistory")
    public ResponseData queryChatHistoryDto(@RequestBody Chathistory chathistory) {
        ResponseData responseData = chathistoryService.queryChatHistoryDto(chathistory.getUid());
        return responseData;
    }

//    @RequestMapping("/chat")
//    public String chatCall(String msg) {
//        return chatClient.prompt()
//                .user(msg)
//                .call()
//                .content();
//    }
//
//    @RequestMapping(value = "/chatString")// produces = "text/html;charset=utf-8"
//    public Flux<String> chatStreamOne(@RequestParam String msg) {
//        Flux<String> content = chatClient.prompt()
//                .user(msg)
//                .stream()
//                .content();
//        return content;
//    }

    @GetMapping(value = "/lang/chat/reactive", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> reactiveChat(@RequestParam("msg") String msg) {
        // 1. 异步调用AI模型
        return Mono.fromCallable(() -> chatLanguageModel.chat(msg))
                .subscribeOn(Schedulers.boundedElastic()) // 在弹性线程池执行阻塞操作
                .timeout(Duration.ofSeconds(30)) // 设置超时时间
                .retry(3) // 重试机制
                .flatMapMany(response -> {
                    // 2. 将响应拆分为5个字符一组
                    AtomicInteger counter = new AtomicInteger(0);
                    int chunkSize = 5;

                    return Flux.range(0, (int) Math.ceil((double) response.length() / chunkSize))
                            .delayElements(Duration.ofMillis(200)) // 控制发送速度
                            .map(i -> {
                                int start = i * chunkSize;
                                int end = Math.min(start + chunkSize, response.length());
                                return response.substring(start, end);
                            })
                            .map(chunk -> {
                                // 3. 构建SSE格式数据
                                return "id: " + counter.getAndIncrement() + "\n" +
                                        "event: message\n" +
                                        "data: " + chunk + "\n\n";
                            })
                            .concatWithValues("event: complete\n" + "data: [DONE]\n\n"); // 结束标志
                })
                .onErrorResume(e -> {
                    // 4. 错误处理
                    return Flux.just(
                            "event: error\n" +
                                    "data: 服务暂时不可用: " + e.getMessage() + "\n\n"
                    );
                });
    }

    @GetMapping(value = "/lang/chat")
    public SseEmitter chat(@RequestParam("msg") String msg) {
        // 设置合理的超时时间（30分钟）
        SseEmitter emitter = new SseEmitter(TimeUnit.MINUTES.toMillis(30));

        executor.execute(() -> {
            try {
                // 1. 调用AI模型获取响应（添加重试机制）
                String message = withRetry(() -> chatLanguageModel.chat(msg), 3, 1000);

                // 2. 分块发送响应
                sendResponseInChunks(emitter, message, 5, 200);

                // 3. 发送完成事件
                emitter.send(SseEmitter.event().name("complete").data("[DONE]"));
                emitter.complete();

            } catch (Exception e) {
                handleError(emitter, e);
            }
        });

        // 4. 添加生命周期回调
        addLifecycleCallbacks(emitter);

        return emitter;
    }

    private void sendResponseInChunks(SseEmitter emitter, String message, int chunkSize, int delayMs)
            throws IOException, InterruptedException {

        for (int i = 0; i < message.length(); i += chunkSize) {

            // 发送数据块
            int endIndex = Math.min(i + chunkSize, message.length());
            String chunk = message.substring(i, endIndex);

            emitter.send(SseEmitter.event()
                    .data(chunk)
                    .id(String.valueOf(i))
                    .name("message")
                    .reconnectTime(3000L));

            // 控制发送速度
            Thread.sleep(delayMs);
        }
    }

    private <T> T withRetry(Supplier<T> supplier, int maxAttempts, long delayMs) throws Exception {
        Exception lastException = null;

        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            try {
                return supplier.get();
            } catch (Exception e) {
                lastException = e;
                if (attempt < maxAttempts) {
                    Thread.sleep(delayMs);
                }
            }
        }

        throw lastException;
    }

    private void handleError(SseEmitter emitter, Exception e) {
        try {
            emitter.send(SseEmitter.event()
                    .name("error")
                    .data("服务暂时不可用，请稍后重试"));
            emitter.completeWithError(e);
        } catch (IOException ex) {
            emitter.completeWithError(ex);
        }
    }

    private void addLifecycleCallbacks(SseEmitter emitter) {
        emitter.onCompletion(() -> log.info("SSE连接正常完成"));
        emitter.onTimeout(() -> log.warn("SSE连接超时"));
        emitter.onError(ex -> log.error("SSE连接出错", ex));
    }

    @RequestMapping(value = "/chatString")// produces = "text/html;charset=utf-8"
    public Flux<String> chatStream(@RequestParam("msg") String msg) {

        Flux<String> message = Flux.create(flux -> {
            streamingChatLanguageModel.chat(msg, new StreamingChatResponseHandler() {

                @Override
                public void onPartialResponse(String s) {
                    flux.next(s);
                }

                @Override
                public void onCompleteResponse(ChatResponse chatResponse) {
                    flux.complete();
                }

                @Override
                public void onError(Throwable throwable) {
                    flux.error(throwable);
                }
            });
        });

        return message;
    }

    @DeleteMapping("/deleteHistoryByRecording/{recording}")
    public void deleteHistoryByRecordingElement(@PathVariable("recording") String recording) {
        chathistoryService.deleteHistoryByRecording(recording);
    }


    /**
     * 连续对话 memoryId 隔离对话，chatLang
     *
     * @param chatMessageHistoryDto
     * @return
     */
    @RequestMapping(value = "/chat")
    public Flux<String> chatLang(@RequestBody ChatMessageHistoryDto chatMessageHistoryDto) {
        System.out.println(chatMessageHistoryDto.getUsername() + ":" + chatMessageHistoryDto.getMemoryId());
        return chatMemoryService.chat(chatMessageHistoryDto.getUsername() + ":" + chatMessageHistoryDto.getMemoryId(), chatMessageHistoryDto.getQuestion());
    }


    @RequestMapping(value = "/chat2")
    public Flux<String> chatLangXLI(@RequestParam String memoryId, @RequestParam String question, @RequestParam String username) {
        String trimMemoryId = memoryId.trim();//解决 memoryId 携带空格字符串
        log.info(username + ":" + trimMemoryId);
        return chatMemoryService.chatGet(username + ":" + trimMemoryId, question);
    }

    /**
     * 获取redis中用户的聊天记录
     *
     * @param username
     * @return
     */
    @RequestMapping("getChatHistoryMessage")
    public ResponseData getChatLangMessage(@RequestParam String username) {
        ResponseData responseData = chathistoryService.getChatLangMessageListByUsername(username);
        return responseData;
    }

    @RequestMapping("deleteConversation")
    public ResponseData deleteConversationChatLangMessage(@RequestParam String sessionId, @RequestParam String username) {
        log.info(sessionId);
        ResponseData responseData = chathistoryService.deleteConversationChatLangMessageBySessionId(sessionId, username);
        return responseData;
    }



    @GetMapping(value = "/stream-text", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamText() {
        // 设置1小时超时（可根据需要调整）
        SseEmitter emitter = new SseEmitter(3600_000L);

        String message = "100个SpringBoot实战项目》将持续推出，不管是学习SpringBoot，还是温习SpringBoot将受益匪浅。本系列将以合集的形式系统化推出，欢迎持续关注。\n" +
                "\n" +
                "共建知识区 你我共同努力。UP推出图文答疑、语音答疑、视频答疑三档增值服务，有需要私信了解详情。";

        executor.execute(() -> {
            try {
                // 逐字发送
                int chunkSize = 5; // 每次发送5个字
                for (int i = 0; i < message.length(); i += chunkSize) {
                    // 截取5个字（防止越界）
                    int endIndex = Math.min(i + chunkSize, message.length());
                    String chunk = message.substring(i, endIndex);
                    // 获取当前字符并确保UTF-8编码
                    String charToSend = String.valueOf(message.charAt(i));

                    // 构造SSE事件
                    emitter.send(SseEmitter.event()
                            .data(chunk)
                            .id(String.valueOf(i)) // 事件ID
                            .name("character")    // 事件名称
                            .reconnectTime(3000L)); // 重连时间

                    // 控制发送速度（100毫秒/字）
                    Thread.sleep(200);

                }

                // 发送完成事件
                emitter.send(SseEmitter.event()
                        .data("[END]")
                        .name("end"));

                // 完成连接
                emitter.complete();
            } catch (IOException | InterruptedException e) {
                // 发生错误时关闭连接
                emitter.completeWithError(e);
            }
        });
        // 连接关闭时的清理操作
        emitter.onCompletion(() -> System.out.println("SSE连接完成"));
        emitter.onTimeout(() -> System.out.println("SSE连接超时"));
        emitter.onError((ex) -> System.out.println("SSE连接错误: " + ex.getMessage()));
        return emitter;
    }
}

